package com.ai.task;

import com.ai.dao.mapper.BfmUserMapper;
import com.ai.service.LoginLogService;
import com.ai.service.OperationLogService;
import com.alibaba.fastjson.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;


/**
 * @Description
 * @ClassName 对接kafka日志审计系统
 * @Author User
 * @date 2023.06.25
 */

@Component
@EnableScheduling
public class KafkaTask implements SchedulingConfigurer {

    @Autowired
    BfmUserMapper bfmUserMapper;

    @Autowired
    private LoginLogService loginLogService;

    @Autowired
    private OperationLogService operationLogService;

    @Autowired
    private KafkaTemplate kafkaTemplate;

    private static int failCount = 0;  // 消息发送失败的数量

    private static int successCount = 0; // 消息发送成功的数量

    private static int i = 0;

    private static KafkaTask kafkaTask = new KafkaTask();

    private boolean condition = false;

    /**
     * kafka的topic
     */

    public static final String TOPIC_LOGIN = "ETE_SECU_ORI_LOG_10069_025";

    public static final String TOPIC_OPER = "ETE_SECU_ORI_LOG_10069_026";

    @Override
    public void configureTasks(ScheduledTaskRegistrar scheduledTaskRegistrar) {
        scheduledTaskRegistrar.addTriggerTask(() -> process(),
                triggerContext -> {
                    String cron = bfmUserMapper.getCron(3);
                    if (cron.isEmpty()) {
                        System.out.println("cron is null");
                    }
                    return new CronTrigger(cron).nextExecutionTime(triggerContext);
                });
    }

    private void process() {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date now = new Date();
        System.out.println("日志推送定时任务开始~准备发送消息");
        // 登陆日志
        List<JSONObject> jsonObjects = loginLogService.getLoginLogs(simpleDateFormat, simpleDateFormat.format(now));
        if (jsonObjects.isEmpty()) {
            System.out.println("10分钟内登陆日志为空");
        } else {
            for (i = 0; KafkaTask.i < jsonObjects.size(); i++) {
                ListenableFuture send = kafkaTemplate.send(TOPIC_LOGIN, jsonObjects.get(i).toJSONString());
                send.addCallback(new ListenableFutureCallback() {
                    @Override
                    public void onFailure(Throwable throwable) {
                        failCount++;
                        synchronized (kafkaTask) {
                            if ((successCount + failCount) == (jsonObjects.size())) {
                                kafkaTask.condition = true;
                                kafkaTask.notifyAll();
                            }
                        }
                    }

                    @Override
                    public void onSuccess(Object o) {
                        successCount++;
                        synchronized (kafkaTask) {
                            if ((successCount + failCount) == (jsonObjects.size())) {
                                kafkaTask.condition = true;
                                kafkaTask.notifyAll();
                            }
                        }
                    }
                });
            }
            if (!jsonObjects.isEmpty()) {
                synchronized (kafkaTask) {
                    while (!kafkaTask.condition) {
                        try {
                            kafkaTask.wait();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
            System.out.println(simpleDateFormat.format(now) + " 登录日志数量总数为：" + jsonObjects.size() + " ,成功发送的数量为："
                    + successCount + " ,发送失败的数量为：" + failCount);
        }
        successCount = 0;
        failCount = 0;
        kafkaTask.condition = false;

        //查询日志
        List<JSONObject> queryLogs = loginLogService.getQueryLogs(simpleDateFormat, simpleDateFormat.format(now));
        if (queryLogs.isEmpty()) {
            System.out.println("10分钟内查询日志为空");
        } else {
            for (i = 0; KafkaTask.i < queryLogs.size(); i++) {
                ListenableFuture send = kafkaTemplate.send(TOPIC_OPER, queryLogs.get(i).toJSONString());
                send.addCallback(new ListenableFutureCallback() {
                    @Override
                    public void onFailure(Throwable throwable) {
                        failCount++;
                        synchronized (kafkaTask) {
                            if ((successCount + failCount) == (queryLogs.size())) {
                                kafkaTask.condition = true;
                                kafkaTask.notifyAll();
                            }
                        }
                    }

                    @Override
                    public void onSuccess(Object o) {
                        successCount++;
                        synchronized (kafkaTask) {
                            if ((successCount + failCount) == (queryLogs.size())) {
                                kafkaTask.condition = true;
                                kafkaTask.notifyAll();
                            }
                        }
                    }
                });
            }
            if (!queryLogs.isEmpty()) {
                synchronized (kafkaTask) {
                    while (!kafkaTask.condition) {
                        try {
                            kafkaTask.wait();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
            System.out.println(simpleDateFormat.format(now) + " 查询日志数量总数为：" + queryLogs.size() + " ,成功发送的数量为："
                    + successCount + " ,发送失败的数量为：" + failCount);
        }
        successCount = 0;
        failCount = 0;
        kafkaTask.condition = false;

        // 操作日志-notebook实例
        List<JSONObject> operationLogs = operationLogService.getOperationLogs(simpleDateFormat, simpleDateFormat.format(now));
        if (operationLogs.isEmpty()) {
            System.out.println("10分钟内操作日志-notebook实例为空");
        } else {
            for (i = 0; i < operationLogs.size(); i++) {
                ListenableFuture send = kafkaTemplate.send(TOPIC_OPER, operationLogs.get(i).toJSONString());
                send.addCallback(new ListenableFutureCallback() {
                    @Override
                    public void onFailure(Throwable throwable) {
                        failCount++;
                        synchronized (kafkaTask) {
                            if ((successCount + failCount) == (operationLogs.size())) {
                                kafkaTask.condition = true;
                                kafkaTask.notifyAll();
                            }
                        }
                    }

                    @Override
                    public void onSuccess(Object o) {
                        successCount++;
                        synchronized (kafkaTask) {
                            if ((successCount + failCount) == (operationLogs.size())) {
                                kafkaTask.condition = true;
                                kafkaTask.notifyAll();
                            }
                        }
                    }
                });
            }
            if (!operationLogs.isEmpty()) {
                synchronized (kafkaTask) {
                    while (!kafkaTask.condition) {
                        try {
                            kafkaTask.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
            System.out.println(simpleDateFormat.format(now) + "notebook操作日志数量总数为：" + operationLogs.size() + " ,成功发送的数量为："
                    + successCount + " ,发送失败的数量为：" + failCount);
        }
        successCount = 0;
        failCount = 0;
        kafkaTask.condition = false;

        // 操作日志-能力管理
        List<JSONObject> operationLogs2 = operationLogService.getOperationLogs2(simpleDateFormat, simpleDateFormat.format(now));
        if (operationLogs2.isEmpty()) {
            System.out.println("10分钟内操作日志-能力管理为空");
        } else {
            for (i = 0; i < operationLogs2.size(); i++) {
                ListenableFuture send = kafkaTemplate.send(TOPIC_OPER, operationLogs2.get(i).toJSONString());
                send.addCallback(new ListenableFutureCallback() {
                    @Override
                    public void onFailure(Throwable throwable) {
                        failCount++;
                        synchronized (kafkaTask) {
                            if ((successCount + failCount) == (operationLogs2.size())) {
                                kafkaTask.condition = true;
                                kafkaTask.notifyAll();
                            }
                        }
                    }

                    @Override
                    public void onSuccess(Object o) {
                        successCount++;
                        synchronized (kafkaTask) {
                            if ((successCount + failCount) == (operationLogs2.size())) {
                                kafkaTask.condition = true;
                                kafkaTask.notifyAll();
                            }
                        }
                    }
                });
            }
            if (!operationLogs2.isEmpty()) {
                synchronized (kafkaTask) {
                    while (!kafkaTask.condition) {
                        try {
                            kafkaTask.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
            System.out.println(simpleDateFormat.format(now) + "能力管理操作日志数量总数为：" + operationLogs2.size() + " ,成功发送的数量为："
                    + successCount + " ,发送失败的数量为：" + failCount);
        }
        successCount = 0;
        failCount = 0;
        kafkaTask.condition = false;

        // 操作日志-能力管理-版本
        List<JSONObject> operationLogs4 = operationLogService.getOperationLogs4(simpleDateFormat, simpleDateFormat.format(now));
        if (operationLogs4.isEmpty()) {
            System.out.println("10分钟内操作日志-能力管理-版本为空");
        } else {
            for (i = 0; i < operationLogs4.size(); i++) {
                ListenableFuture send = kafkaTemplate.send(TOPIC_OPER, operationLogs4.get(i).toJSONString());
                send.addCallback(new ListenableFutureCallback() {
                    @Override
                    public void onFailure(Throwable throwable) {
                        failCount++;
                        synchronized (kafkaTask) {
                            if ((successCount + failCount) == (operationLogs4.size())) {
                                kafkaTask.condition = true;
                                kafkaTask.notifyAll();
                            }
                        }
                    }

                    @Override
                    public void onSuccess(Object o) {
                        successCount++;
                        synchronized (kafkaTask) {
                            if ((successCount + failCount) == (operationLogs4.size())) {
                                kafkaTask.condition = true;
                                kafkaTask.notifyAll();
                            }
                        }
                    }
                });
            }
            if (!operationLogs4.isEmpty()) {
                synchronized (kafkaTask) {
                    while (!kafkaTask.condition) {
                        try {
                            kafkaTask.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
            System.out.println(simpleDateFormat.format(now) + "能力管理-版本操作日志数量总数为：" + operationLogs4.size() + " ,成功发送的数量为："
                    + successCount + " ,发送失败的数量为：" + failCount);
        }
        successCount = 0;
        failCount = 0;
        kafkaTask.condition = false;

        // 操作日志-标注任务
        List<JSONObject> operationLogs3 = operationLogService.getOperationLogs3(simpleDateFormat, simpleDateFormat.format(now));
        if (operationLogs3.isEmpty()) {
            System.out.println("10分钟内操作日志-标注任务为空");
        } else {
            for (i = 0; i < operationLogs3.size(); i++) {
                ListenableFuture send = kafkaTemplate.send(TOPIC_OPER, operationLogs3.get(i).toJSONString());
                send.addCallback(new ListenableFutureCallback() {
                    @Override
                    public void onFailure(Throwable throwable) {
                        failCount++;
                        synchronized (kafkaTask) {
                            if ((successCount + failCount) == (operationLogs3.size())) {
                                kafkaTask.condition = true;
                                kafkaTask.notifyAll();
                            }
                        }
                    }

                    @Override
                    public void onSuccess(Object o) {
                        successCount++;
                        synchronized (kafkaTask) {
                            if ((successCount + failCount) == (operationLogs3.size())) {
                                kafkaTask.condition = true;
                                kafkaTask.notifyAll();
                            }
                        }
                    }
                });
            }
            if (!operationLogs3.isEmpty()) {
                synchronized (kafkaTask) {
                    while (!kafkaTask.condition) {
                        try {
                            kafkaTask.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
            System.out.println(simpleDateFormat.format(now) + "标注任务操作日志数量总数为：" + operationLogs3.size() + " ,成功发送的数量为："
                    + successCount + " ,发送失败的数量为：" + failCount);
        }
        System.out.println("日志推送定时任务结束~");
    }
}

